<?php
declare(strict_types=1);

namespace ServerApp\lib\server\delay_push;

use Exception;
use ServerApp\lib\process\SProcess;
use ServerApp\lib\util\ServerHelper;
use Tao\src\model\mysql\tao\PushMessageDelay;
use Throwable;
use WebApp\lib\util\Util;

/**
 * Class DelayPush
 * @package txz\client\push\server\delay_push
 */
class DelayPush
{
    private $once_rows = 1000;
    private $next_rows = [];
    private $mid = 0;

    /**
     * @var SProcess
     */
    private $process;
    private $process_id;

    private $pushTask = [];
    private $pushTaskNum = 3;
    private $pushTaskUsing = [];

    private function createPushTask(){
        $taskProcess = new SProcess(new Task($this->func), false, 2, 'task_push');
        $pid = $taskProcess->start();
        $this->pushTask[$pid] = $taskProcess;
        $this->pushTaskUsing[$pid] = false;
        swoole_event_add($taskProcess->pipe, function () use ($taskProcess) {
            $this->pushTaskUsing[(int)trim($taskProcess->read())] = false;
        });
    }

    private function createReadTask(){
        $this->process = new SProcess(new HandleDb(), false, 2, 'task_read');
        swoole_event_add($this->process->pipe, function(){
            $this->HandleDbEvent();
        });
        $this->process_id = $this->process->start();
    }

    private function writePid(){
        $pid[] = $this->process_id;
        foreach ($this->pushTask as $Key => $val){
            $pid[] = $Key;
        }
        $content = implode('|', $pid);
        $s = explode('\\', __CLASS__);
        file_put_contents(ServerHelper::getInstance()->getPidPath() . end($s) . __LINE__ . ".pid", $content);
    }

    public $func;
    /**
     * DelayPush constructor.
     * @param IDelayPush $func2
     */
    private function __construct(IDelayPush $func2){
        $this->func = $func2;
        $name = 'task_main';
        tracker(D, "$name : ".posix_getpid());
        swoole_set_process_name($name);
        for ($i = 0; $i < $this->pushTaskNum; $i++) {
            $this->createPushTask();
        }

        $this->createReadTask();

        swoole_timer_tick(200,function(){
            $this->send_rows_to_task();
        });

        $this->writePid();

        SProcess::signal(SIGCHLD, function ($sig) {
            tracker(D, "sig:$sig");
            for(;;){
                $ret = SProcess::wait();
                if($ret){
                    tracker(D, $ret);
                    if ($this->process_id == $ret['pid']){
                        $this->createReadTask();
                        $this->writePid();
                    }else{
                        if (isset($this->pushTask[$ret['pid']])){
                            unset($this->pushTask[$ret['pid']]);
                            $this->createPushTask();
                            $this->writePid();
                        }
                    }
                } else {
                    tracker(D, $ret);
                }
            }
        });
    }

    private static $instance = null;

    /**
     * @param IDelayPush $func2
     */
    public static function init(IDelayPush $func2) {
        if (self::$instance === null) {
            self::$instance = new self($func2);
        }
    }

    public function HandleDbEvent(){
        $key = $this->process->read();
        if (strlen($key) > 10) {
            try{
                $rows = json_decode(file_get_contents($key), true);
                foreach ($rows as $row) {
                    $this->next_rows[] = $row;
                }
                if (isset($row)){
                    if (count($rows) < HandleDb::READ_ROWS){
                        $this->mid = 0;
                        $this->process->write(HandleDb::SLEEP);
                        return;
                    }else{
                        $this->mid = $row[PushMessageDelay::MID];
                    }

                    if (count($this->next_rows) > 20000){
                        $this->process->write(HandleDb::WAIT);
                        return;
                    }
                }
            }catch(\Throwable $e){
                Util::err($e);
            }
        }
        $this->process->write($this->mid.'');
    }

    private function send_rows_to_task(){
        try{
            if ($this->next_rows) {
                $break = false;
                foreach ($this->pushTaskUsing as $pid => $bool) {
                    if (!$bool) {
                        if (count($this->next_rows) > $this->once_rows) {
                            $data = array_splice($this->next_rows, 0, $this->once_rows);
                        } else {
                            $data = $this->next_rows;
                            $this->next_rows = [];
                        }

                        if ($data) {
                            $key = __DIR__ . '/task_rows';
                            file_put_contents($key, json_encode($data));
                            $this->pushTask[$pid]->write($key);
                            $this->pushTaskUsing[$pid] = true;
                        }

                        $break = $pid;
                        break;
                    }
                }
                if ($break === false) {
                    echo "no use process\n";
                }else{
                    $this->send_rows_to_task();
                }
            }
        } catch (Exception $e) {
            Util::err($e);
        }catch (Throwable $e){
            Util::err($e);
        }
    }
}

//SProcess::daemon(true, true);
//php7 /data/web/server/client/push/server/delay_push/DelayPush.php
//ps -ef | grep DelayPush.php | grep -v grep | cut -c 9-15 | xargs kill -s 9
//ps -ef | grep python3 | grep -v grep | cut -c 9-15 | xargs kill -s 9
//ps -ef | grep swoole_worker_main | grep -v grep | cut -c 9-15 | xargs kill -s 9;python2 /data/web/server/client/base/check_swoole.py
//ps -ef | grep /data/web/server/oss/tools/audio_spider/qqmusic_cp/new_all/GetMonth_qqmusic.py | grep -v grep | cut -c 9-15 | xargs kill -s 9
//ps -ef | grep /data/web/server/oss/tools/audio_spider/qqmusic_cp/Get_gedan.py | grep -v grep | cut -c 9-15 | xargs kill -s 9
///data/web/server/test/test_conn/let_user_logout.php 1585
